package cc.example;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
import reactor.core.scheduler.Schedulers;
import reactor.util.context.Context;

import javax.swing.*;
import java.time.Duration;
import java.util.concurrent.LinkedBlockingDeque;

public class SinksDemo {
    public static void main(String[] args) throws InterruptedException {
//        Flux.create();

        Sinks.many().unicast(); // 单播    这个管道只能绑定单个订阅者（消费者）
        Sinks.many().multicast(); // 多播  这个管道能绑定多个订阅者
        Sinks.many().replay(); // 重放    这个管道能重放元素 是否给后来的订阅者把之前的元素依然发给它；


        Sinks.Many<Object> many = Sinks.many().unicast().onBackpressureBuffer(
                new LinkedBlockingDeque<>(5)
        );

        new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                many.tryEmitNext("a" + i);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }).start();

        many.asFlux().subscribe(o -> {
            System.out.println(o);
        });
        many.asFlux().subscribe(o -> {
            System.out.println(o);
        });



        Flux.range(1,10)
                .cache(3); // 缓存元素

        Flux.range(1, 10)
                .collectList()
                .block(); // 也是一种订阅者



        Flux.range(1, 100)
                .buffer(10)
                .parallel(8)
                .runOn(Schedulers.newParallel("dd"))
                .log()
                .subscribe(integers -> {
                    System.out.println(integers);
                });


        Flux.just(1,2,3)
                .transformDeferredContextual((integerFlux, contextView) -> {
                    return Flux.just(2,3,4);
                })
                .map(i -> {
                    System.out.println("i=" + i);
                    return i+10;
                })

                //上游能拿到下游的最近一次数据
                .contextWrite(Context.of("prefix", "haha"))
                //ThreadLocal共享了数据，上游的所有人能看到；Context由下游传播给上游
                .subscribe(integer -> {
                    System.out.println("v=" + integer);
                });

    }
}
